package sanbing.jcpplearning;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;

public class MQTTServer {
    public static void main(String[] args) throws InterruptedException {
        // 线程组：bossGroup 处理连接，workerGroup 处理 I/O
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new MqttDecoder());  // MQTT 解码器
                            ch.pipeline().addLast(MqttEncoder.INSTANCE);  // MQTT 编码器
                            ch.pipeline().addLast(new MQTTServerHandler()); // 自定义处理器
                        }
                    });

            // 绑定端口并启动服务
            System.out.println("MQTT Server is running on port 1883...");
            ChannelFuture future = bootstrap.bind(1883).sync(); // 绑定端口 1883
            future.channel().closeFuture().sync(); // 等待关闭
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

// 自定义处理器：处理 MQTT 消息
class MQTTServerHandler extends SimpleChannelInboundHandler<MqttMessage> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
        // 打印接收到的 MQTT 消息类型
        System.out.println("Server received message: " + msg.fixedHeader().messageType());

        if (msg.fixedHeader().messageType() == MqttMessageType.CONNECT) {
            System.out.println("Received CONNECT message from client");

            // 回复 CONNACK 消息
            MqttConnAckMessage connAckMessage = MqttMessageBuilders.connAck()
                    .returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
                    .build();
            ctx.writeAndFlush(connAckMessage); // 发送 CONNACK 消息

        } else if (msg.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
            // 处理 PUBLISH 消息
            handlePublishMessage(ctx, (MqttPublishMessage) msg);

        } else if (msg.fixedHeader().messageType() == MqttMessageType.PUBREL) {
            // 处理 PUBREL（QoS 2）
            handlePubRelMessage(ctx, msg);

        } else if (msg.fixedHeader().messageType() == MqttMessageType.PINGREQ) {
            // 构造并发送 PINGRESP 消息
            MqttMessage pingResp = new MqttMessage(
                new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0)
            );
            ctx.writeAndFlush(pingResp); // 发送 PINGRESP 消息

        } else if (msg.fixedHeader().messageType() == MqttMessageType.DISCONNECT) {
            System.out.println("Client disconnected");
            ctx.close(); // 关闭连接
        } else {
            System.out.println("Unsupported message type: " + msg.fixedHeader().messageType());
        }
    }

    private void handlePublishMessage(ChannelHandlerContext ctx, MqttPublishMessage publishMessage) {
        // 获取 Topic
        String topicName = publishMessage.variableHeader().topicName();

        // 获取 Payload
        String payload = publishMessage.payload().toString(CharsetUtil.UTF_8);

        // 打印 Topic 和 Payload
        System.out.println("Received PUBLISH message:");
        System.out.println("  Topic: " + topicName);
        System.out.println("  Payload: " + payload);

        // 获取 QoS 等级
        MqttQoS qos = publishMessage.fixedHeader().qosLevel();

        if (qos == MqttQoS.AT_LEAST_ONCE) {
            // 回复 PUBACK（QoS 1）
            int packetId = publishMessage.variableHeader().packetId();
            MqttMessage pubAckMessage = MqttMessageBuilders.pubAck()
                    .packetId(packetId)
                    .build();
            ctx.writeAndFlush(pubAckMessage);

        } else if (qos == MqttQoS.EXACTLY_ONCE) {
            // 回复 PUBREC（QoS 2 第一步）
            int packetId = publishMessage.variableHeader().packetId();
            MqttMessage pubRecMessage = new MqttMessage(
                    new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                    MqttMessageIdVariableHeader.from(packetId)
            );
            ctx.writeAndFlush(pubRecMessage);
        }
    }

    private void handlePubRelMessage(ChannelHandlerContext ctx, MqttMessage pubRelMessage) {
        // 获取 Packet ID
        int packetId = ((MqttMessageIdVariableHeader) pubRelMessage.variableHeader()).messageId();

        // 回复 PUBCOMP（QoS 2 最后一步）
        MqttMessage pubCompMessage = new MqttMessage(
                new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(packetId)
        );
        ctx.writeAndFlush(pubCompMessage);

        System.out.println("Completed QoS 2 message exchange for packet id: " + packetId);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}